package com.xxx.consumer;

import com.alibaba.fastjson.JSONObject;
import com.xxx.domain.MessageStation;
import com.xxx.dto.StationMessage2MqDto;
import com.xxx.service.IMessageSmsService;
import com.xxx.service.IMessageStationService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Date;

@RocketMQMessageListener(
        consumerGroup = "station-message-consumer"
        , topic = "topic-message"
        , selectorExpression = "station"
        , messageModel = MessageModel.CLUSTERING
)
@Component
@Slf4j
public class SationMessageConsumer implements RocketMQListener<MessageExt> {

    @Autowired
    private IMessageStationService messageStationService;

    @Override
    public void onMessage(MessageExt message) {

        byte[] body = message.getBody();
        String messageDtoJsonStr = new String(body, StandardCharsets.UTF_8);
        StationMessage2MqDto stationMessage2MqDto = JSONObject.parseObject(messageDtoJsonStr,
                StationMessage2MqDto.class);

        log.info("接收到消息:{}"+messageDtoJsonStr);

        for (Long userId : stationMessage2MqDto.getUserIds()) {
            MessageStation messageStation = new MessageStation();
            BeanUtils.copyProperties(stationMessage2MqDto, messageStation);

            messageStation.setUserId(userId);
            messageStation.setIsread(0);
            messageStation.setSendTime(new Date());
            messageStationService.insert(messageStation);
        }

    }
}
